package com.wanke.subscribe;

import com.wanke.config.MeasurementCacheConfig;
import com.wanke.constant.EventConstant;
import com.wanke.domain.po.Measurement;
import com.wanke.domain.po.Measurements;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;


public class MeasureTransmitterSubscribe extends AbstractVerticle {


    private int pollSize;

    private int timeOut;

    private int periodic;

    private Logger logger = LoggerFactory.getLogger(MeasureTransmitterSubscribe.class);

    MeasurementCacheConfig measurementCacheConfig = MeasurementCacheConfig.INSTANCE;


    public MeasureTransmitterSubscribe(int pollSize, int timeOut, int periodic) {
        this.pollSize = pollSize;
        this.timeOut = timeOut;
        this.periodic = periodic;
    }

    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        vertx.setPeriodic(periodic, id -> {
            //测点发送队列不为空，发送数据
            if (measurementCacheConfig.getSize() > 0) {
                List<Measurement> cloudMeasureValues;
                Measurement currMeasureValue = measurementCacheConfig.peek();
                if ((currMeasureValue.getFlag() & 0x01) > 0) { // 5秒跑一次，取出头部第一个元素。只有实时测点会放在头部，非实时放在尾部
                    //存在实时测点,直接全量发送
                    cloudMeasureValues = measurementCacheConfig.extractCloudMeasurement(measurementCacheConfig.getSize());
                    logger.debug("实时测点数据消息[{}]", cloudMeasureValues.size());
                    Measurements objects = new Measurements();
                    objects.setMeasurementList(cloudMeasureValues);
                    this.vertx.eventBus().publish(EventConstant.MEAS_PUBLISH.name(), objects);
                } else {
                    //没有实时测点，数据延时或数据数量足够后，打包后发送
                    long diffTime = System.currentTimeMillis() - currMeasureValue.getStoreTime().getTime();
                    if (diffTime > 3 * 1000 || measurementCacheConfig.getSize() > pollSize) {
                        //延时5秒，或 数量超过500条，生成一条云端消息
                        cloudMeasureValues = measurementCacheConfig.extractCloudMeasurement(measurementCacheConfig.getSize());
                        logger.debug("常规测点数据消息[{}]", cloudMeasureValues.size());
                        Measurements objects = new Measurements();
                        objects.setMeasurementList(cloudMeasureValues);
                        this.vertx.eventBus().publish(EventConstant.MEAS_PUBLISH.name(), objects);
                    }
                }
            }
        });
    }

}
